FastAPIとSQLAlchemy によるWebサービスの実装

FastAPI から SQLAlchemy を使ってデータベースを利用したRESTful Webサービスを実装してみます。

SQLAlchemy は単独でFastAPIから利用することができますが、より便利な拡張機能も公開されています。

fastapi-restful

fastapi-utils

databases

ormantic

HTTPメソッドのマッピング

WebサービスTODOの仕様を少し変更しています。

理解が容易になる事を重視して認証によるWebサービスの保護は省いています。

table: APIとHTTPメソッド

HTTPメソッド URI アクション

タスクリソースは次の情報を持つものとします。

SQLAlchemy による実装

まず、SQLAlchemyで実装してみます。

インストール

code: bash

$ pip install alembic

Python 3.7 でなければ次のモジュールもインストールしておきます。

code: bash

$ pip install async-exit-stack async-generator

準備

アプリケーションのディレクトリを作成します。

code: bash

$ mkdir -p $HOME/fastapi/todo_apiv2

$ cd $HOME/fastapi/todo_apiv2

SQLAlchemy と Pydantic のスタイルの違い

既に説明したように FastAPI は pydantic のデータ検証をベースにして開発されています。

ここで、SQLAlchemy と Pydantic ではフィールド定義の方法が違うことを理解しておく必要があります。

code: python

name = Column(String, default='python')

code: python

name: str = 'python'

データベースの初期設定

code: python

import sqlalchemy as sa

from sqlalchemy.orm import sessionmaker

db_engine = sa.create_engine('sqlite:///todo.db')

Session = sessionmaker(autocommit=False,

autoflush=False,

bind=db_engine)

def get_db():

db = Session()

try:

yield db

finally:

db.close()

SQLAlchemyのモデルクラス

code: python

import sqlalchemy as sa

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Task(Base):

__tablename__ = 'tasks'

id = sa.Column(sa.Integer, primary_key = True)

title = sa.Column(sa.String(32), index = True)

description = sa.Column(sa.Text(128), nullable=True)

done = sa.Column(sa.Boolean)

Alembic でデータベースを作成

モデルクラスを使って自分でデータベースを作成してもよいのですが、

ここではAlembic でデータベースを作成しましょう。

code: bash

$ alembic init migrations

code: python

sqlalchemy.url = sqlite:///todo.db

code: python

from models import Base

target_metadata = Base.metadata

これでデータベースを作成することができるようになります。

code: bash

$ export PYTHONPATH=.

$ alembic revision --autogenerate -m "db initialize"

$ alembic upgrade head

$ alembic stamp head

Pydantic のモデルクラス

Pydanticモデルは検証のためのスキーマを定義します。

code: python

from pydantic import BaseModel

class TaskSchema(BaseModel):

title: str = Field(...,

title='The title of task',

max_length=32)

description: str = Field(None,

title='The title of task',

max_length=128)

done: bool = Field(False, title='The status of task')

class config:

orm_mode = True

CRUD操作

データベースにアクセスするためのCRUD操作を再利用可能なモジュールにしておきます。

ファイル名: crud.py

code: python

from sqlalchemy.orm import Session

import models

import schemas

def get_task(db: Session, task_id: int):

return db.query(models.Task).filter(models.Task.id == task_id).first()

def get_tasklist(db: Session, skip: int = 0, limit: int = 100):

return db.query(models.Task).offset(skip).limit(limit).all()

def create_task(db: Session, task: schemas.TaskSchema):

db_task = models.Task(title=task.title,

description=task.description,

done=task.done)

db.add(db_task)

db.commit()

db.refresh(db_task)

return db_task

def update_task(db: Session, task_id: int, task: schemas.TaskSchema):

db_task = get_task(db, task_id)

if db_task is not None:

db_task.title = task.tiel

db_task.description = task.description

db_task.done = task.done

db.commit()

db.refresh(db_task)

return db_task

def delete_task(db: Session, task_id: int):

db_task = get_task(db, task_id)

if db_task is not None:

db.delete(db_task)

db.commit()

db.refresh(db_task)

return db_task

FastAPI app.py

RESTful Webアプリケーションで、データベースを処理するときはリクエストごとに独立したデータベースセッションが必要で、すべてのリクエストで同じセッションを使用して、リクエストされた処理が完了したらそれを閉じます。

そして、次のリクエストのために新しいセッションが作成します。

code: python

from fastapi import Depends, FastAPI, HTTPException, Path

from pydantic import BaseModel

from sqlalchemy.orm import sessionmaker, Session

from models import Base

from database import db_engine, get_db

import models

import schemas

import crud

app = FastAPI()

@app.get("/todo/api/v2.0/tasks/")

def get_tasklist(skip: int = 0,

limit: int = 100,

db: Session = Depends(get_db)):

tasklist = crud.get_tasklist(db, skip=skip, limit=limit)

return tasklist

@app.get("/todo/api/v2.0/tasks/{id}")

def get_task(id: int, db: Session = Depends(get_db)):

db_task = crud.get_task(db, task_id=id)

if db_task is None:

raise HTTPException(status_code=404, detail="Task not found")

return db_task

@app.post("/todo/api/v2.0/tasks")

def create_task(task: schemas.TaskSchema,

db: Session = Depends(get_db)):

return crud.create_task(db=db, task=task)

@app.patch("/todo/api/v2.0/tasks/{id}")

def patch_task(id: int,

task: schemas.TaskSchema,

db: Session = Depends(get_db)):

db_task = crud.update_task(db, task_id=id, task=taask)

return db_task

@app.delete("/todo/api/v2.0/tasks/{id}")

def delete_task(id: int, db: Session = Depends(get_db)) :

db_task = crud.delete_task(db, task_id=id)

if db_task is None:

raise HTTPException(status_code=404, detail="Task not found")

return {"result": "OK"}

レスポンスデータの検証

レスポンスデータの検証を行わせたいときがあります。

この場合は、次のように検証するスキーマを与えます。

code: python

from fastapi import FastAPI

import schemas

# ...

app = FastAPI()

@app.get("/todo/api/v2.0/tasks/{id}", response_model=schema.TaskSchema)

def get_task(id: int, db: Session = Depends(get_db)):

db_task = crud.get_task(db, task_id=id)

if db_task is None:

raise HTTPException(status_code=404, detail="Task not found")

return db_task

このときも、同様の問題が発生することがあります。

pydantic では、記法の違いを吸収するための機能が提供されていて、

独自のコードでモデルインスタンスを初期化するときはsnake_case属性名を使用しますが、外部からのリクエストではcamelCase属性を受け入れるように定義することができます。

さらに、レスポンスデータを辞書型に変換できないけれど、適切に名前が付けられたフィールドを持つオブジェクトを返す場合も、インターナルエラーとなってしまいます。

code: bash

pydantic.error_wrappers.ValidationError: 1 validation error for TaskSchema

response

value is not a valid dict (type=type_error.dict)

fastapi-utils や fastapi-restful を利用するとAPIModel を継承したスキームを定義することで、こうした問題をうまく処理してくれます。

両方とも良く似た機能を提供していますが、fastapi-restful の方が後発ということもあり、若干機能が豊富です。

FastAPI-Utils を使った実装

インストール

fastapi-utils の拡張モジュールをインストールします。

code: bash

$ pip install fastapi-utils

APIModel

code: python

from pydantic import Field

from fastapi_utils.api_model import APIModel

class TaskSchema(APIModel):

id: int = Field(None)

title: str = Field(...,

title='The title of task',

max_length=32)

description: str = Field(None,

title='The description of task',

max_length=128)

done: bool = Field(False, title='The status of task')

class config:

orm_mode = True

クラスベースビュー

エンドポイントを追加するAPIRouterを作成します

共有する依存関係ごとに、Dependsタイプの値を持つクラス属性を追加します

元の「非共有」依存関係の使用を、self.dependency のアクセスに置き換えます

code: python

from typing import List

from fastapi import Depends, FastAPI, HTTPException, Path

from sqlalchemy.orm import Session

from fastapi_utils.cbv import cbv

from fastapi_utils.inferring_router import InferringRouter

from database import get_db

import models

import schemas

import crud

app = FastAPI()

router = InferringRouter()

@cbv(router)

class TaskCBV:

db: Session = Depends(get_db)

@router.get("/todo/api/v2.0/tasks/")

def get_tasklist(self, skip: int = 0, limit: int = 100):

tasklist = crud.get_tasklist(self.db, skip=skip, limit=limit)

return tasklist

@router.get("/todo/api/v2.0/tasks/{id}")

def get_task(self, id: int):

db_task = crud.get_task(self.db, task_id=id)

if db_task is None:

raise HTTPException(status_code=404, detail="Task not found")

return db_task

@router.post("/todo/api/v2.0/tasks")

def create_task(self, task: schemas.TaskSchema):

return crud.create_task(self.db, task=task)

@router.patch("/todo/api/v2.0/tasks/{id}")

def patch_task(self, id: int, task: schemas.TaskSchema):

db_task = crud.update_task(db, task_id=id, task=taask)

return db_task

@router.delete("/todo/api/v2.0/tasks/{id}")

def delete_task(self, id: int):

db_task = crud.delete_task(self.db, task_id=id)

if db_task is None:

raise HTTPException(status_code=404, detail="Task not found")

return {"result": "OK"}

app.include_router(router)

クラス変数に１度だけ定義して、それを参照するだけなので記述が少なくなります。

FastAPI-RESTful を使った実装

インストール

fastapi-restful の拡張モジュールをインストールします。

Python3.6以降と pydantic 1.0以降が必要になります。

code: bash

$ pip install fastapi-restful

クラスベースビュー

code: python

from fastapi_restful.cbv import cbv

from fastapi_restful.inferring_router import InferringRouter

Resourceクラス

FastAPI-RESTful の Resourceクラスは、FastAPI-Utils にはない機能です。これを利用すると、APIの機能とリソースを結びつけ、オブジェクト志向プログラミングをサポートするCRUDアプリケーションを簡単に作成することができるようになります。

code: python

from fastapi import FastAPI

from fastapi_restful import Api, Resource

class MyApi(Resource):

def get(self):

return "done"

def main():

app = FastAPI()

api = Api(app)

myapi = MyApi()

api.add_resource(myapi, "/uri")

if __name__ == "__main__":

main()

依存関係が増えるような場合でも、Resourceクラスを継承するクラスのコンストラクタに定義を追加するだけですみます。

また、応答ボディーも pydantic の BaseModel 継承して簡潔に定義することができます。

code: python

from fastapi import FastAPI

from pydantic import BaseModel

from pymongo import MongoClient

from fastapi_restful import API, Resource, set_responses

class ResponseModel(BaseModel):

answer: str

class NotFoundModel(BaseModel):

IsFound: bool

class MyApi(Resource):

def __init__(self, mongo_client):

self.mongo = mongo_client

@set_responses(ResponseModel)

def get(self):

return "done"

@set_responses(ResponseModel, 201, {404: NotFoundModel})

def post(self):

return "Done again"

def main():

app = FastAPI()

api = Api(app)

mongo_client = MongoClient("mongodb://localhost:27017")

myapi = MyApi(mongo_client)

api.add_resource(myapi, "/uri")

if __name__ == "__main__":

main()

APIModel

pydantic の BaseModel の代わりに APIModel を利用するとフィールド定義が簡潔になります。

code: python

from dataclasses import dataclass

from typing import NewType

from fastapi import FastAPI

from fastapi_restful.api_model import APIModel

class TaskSchema(APIModel):

id: int = Field(None)

title: str = Field(...,

title='The title of task',

max_length=32)

description: str = Field(None,

title='The description of task',

max_length=128)

done: bool = Field(False, title='The status of task')

@dataclass

class TaskORM:

id: int

title: str

description: str

done: bool

app = FastAPI()

@app.post("/tasks", response_model=TaskSchema)

def create_user(task: TaskSchema) -> TaskORM:

return TaskORM(task.id, task.title, task.description)

TaskORM が SQLAlchemy のモデルクラスのように振る舞ってくれます。